[DRAFT] Switch to Connexion 3 framework #39055

potiuk · 2024-04-16T08:44:06Z

This is a huge PR being result of over a 100 commits made by a number of people in ##36052 and #37638. It switches to Connexion 3 as the driving backend
implementation for both - Airflow REST APIs and Flask app that powers Airflow UI. It should be largely
backwards compatible when it comes to behaviour of both APIs and Airflow Webserver views, however due to decisions made by Connexion 3 maintainers, it changes heavily the technology stack used under-the-hood:

Connexion App is an ASGI-compatible Open-API spec-first
framework using ASGI as an interface between webserver
and Python web application. ASGI is an asynchronous
successor of WSGI.
Connexion itself is using Starlette to run asynchronous
web services in Python.
We continue using gunicorn appliation server that still
uses WSGI standard, which means that we can continue using
Flask and we are usig standard Uvicorn ASGI webserver that
converts the ASGI interface to WSGI interface of Gunicorn

Some of the problems handled in this PR

There were two problem was with session handling:

the get_session_cookie - did not get the right cookie - it returned "session" string. The right fix was to change cookie_jar into cookie.jar because this is where apparently TestClient of starlette is holding the cookies (visible when you debug)
The client does not accept "set_cookie" method - it accepts passing cookies via "cookies" dictionary - this is the usual httpx client
- see https://www.starlette.io/testclient/ - so we have to set cookie directly in the get method to try it out

Add "flask_client_with_login" for tests that neeed flask client

Some tests require functionality not available to Starlette test client as they use Flask test client specific features - for those we have an option to get flask test client instead of starlette one.

Fix error handling for new connection 3 approach

Error handling for Connexion 3 integration needed to be reworked.

The way it behaves is much the same as it works in main:

for API errors - we get application/problem+json responses
for UI erros - we have rendered views
for redirection - we have correct location header (it's been missing)
the api error handled was not added as available middleware in the www tests

It should fix all test_views_base.py tests which were failing on lack of location header for redirection.

Fix wrong response is tests_view_cluster_activity

The problem in the test was that Starlette Test Client opens a new connection and start new session, while flask test client uses the same database session. The test did not show data because the data was not committed and session was not closed - which also failed sqlite local tests with "database is locked" error.

Fix test_extra_links

The tests were failing again because the dagrun created was not committed and session not closed. This worked with flask client that used the same session accidentally but did not work with test client from Starlette. Also it caused "database locked" in sqlite / local tests.

Switch to non-deprecated auth manager

Fix to test_views_log.py

This PR partially fixes sessions and request parameter for test_views_log. Some tests are still failing but for different reasons - to be investigated.

Fix views_custom_user_views tests

The problem in those tests was that the check in security manager was based on the assumption that the security manager was shared between the client and test flask application - because they were coming from the same flask app. But when we use starlette, the call goes to a new process started and the user is deleted in the database - so the shortcut of checking the security manager did not work.

The change is that we are now checking if the user is deleted by calling /users/show (we need a new users READ permission for that)

this way we go to the database and check if the user was indeed deleted.

Fix test_task_instance_endpoint tests

There were two reasons for the test failed:

when the Job was added to task instance, the task instance was not merged in session, which means that commit did not store the added Job
some of the tests were expecting a call with specific session and they failed because session was different. Replacing the session with mock.ANY tells pytest that this parameter can be anything - we will have different session when when the call will be made with ASGI/Starlette

Fix parameter validation

added default value for limit parameter across the board. Connexion 3 does not like if the parameter had no default and we had not provided one - even if our custom decorated was adding it. Adding default value and updating our decorator to treat None as default fixed a number of problems where limits were not passed
swapped openapi specification for /datasets/{uri} and /dataset/events. Since {uri} was defined first, connection matched events with {uri} and chose parameter definitions from {uri} not events

Fix test_log_enpoint tests

The problem here was that some sessions should be committed/closed but also in order to run it standalone we wanted to create log templates in the database - as it relied implcitly on log templates created by other tests.

Fix test_views_dagrun, test_views_tasks and test_views_log

Fixed by switching to use flask client for testing rather than starlette. Starlette client in this case has some side effects that are also impacting Sqlite's session being created in a different thread and deleted with close_all_sessions fixture.

Fix test_views_dagrun

Fixed by switching to use flask client for testing rather than starlette. Starlette client in this case has some side effects that are also impacting Sqlite's session being created in a different thread and deleted with close_all_sessions fixture.

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

potiuk · 2024-04-16T08:46:24Z

.github/workflows/basic-tests.yml

@@ -148,6 +148,8 @@ jobs:
        env:
          HATCH_ENV: "test"
        working-directory: ./clients/python
+      - name: Compile www assets


We need to have assets compiled here to test Python client. Right now the API uses UI from Swagger that expects Javascript to be compiled in order to make API calls.

potiuk · 2024-04-16T08:50:42Z

airflow/api_connexion/endpoints/connection_endpoint.py

@@ -91,7 +91,7 @@ def get_connection(*, connection_id: str, session: Session = NEW_SESSION) -> API
 @provide_session
 def get_connections(
    *,
-    limit: int,
+    limit: int | None = None,


@RobbeSneyders -> we found that the new Connexion 3 implements more validation and responds with error where no default values are provided for parameters that are no defined as optional (but they are provided by decorators - see above @format_parameters decorator - it will set the default value of the parameter if not set - but Connexion 3 will respond with "bad request" if the parameter has no default value. We fixed it by adding default None and reworking the decorator to check it and replace the default value (and apply range limits) if None is set, but I just wanted you to know and ask if that is intended behaviour/right way of fixing it.

I can't immediately think of a change that could trigger this. I tried to reproduce it with a minimal example, but it works as expected:

def insert(**kwargs): def decorator(f): async def wrapped_function(): return await f(**kwargs) return wrapped_function return decorator @insert(name="Igor") async def post_greeting(name: str): return f"Hello {name}", 200

paths: /greeting: post: operationId: hello.post_greeting parameters: - name: name in: query schema: type: string

Do you have a stack trace by any chance?

If you want to reduce the complexity of this PR, could such simple bulk changes separated-out to s different PR and merged beforehand? Or would this influence/break the old Connexion setup?

We could do that, yes @jscheffl eventually, yes but I would like to avoid "polluting" current code without knowing how we are proceeding with the whole PR. For me this is more of POC for comments and deciding what's next rather than something we want to actively start merging now

@RobbeSneyders: yes:

Example stack trace here (from https://github.com/apache/airflow/actions/runs/8581603105/job/23518857077 for example).

ERROR connexion.middleware.exceptions:exceptions.py:97 TypeError("get_dataset_events() missing 1 required keyword-only argument: 'limit'") Traceback (most recent call last): File "/usr/local/lib/python3.8/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app await app(scope, receive, sender) File "/usr/local/lib/python3.8/site-packages/connexion/middleware/swagger_ui.py", line 222, in __call__ await self.router(scope, receive, send) File "/usr/local/lib/python3.8/site-packages/starlette/routing.py", line 756, in __call__ await self.middleware_stack(scope, receive, send) File "/usr/local/lib/python3.8/site-packages/starlette/routing.py", line 776, in app await route.handle(scope, receive, send) File "/usr/local/lib/python3.8/site-packages/starlette/routing.py", line 485, in handle await self.app(scope, receive, send) File "/usr/local/lib/python3.8/site-packages/starlette/routing.py", line 756, in __call__ await self.middleware_stack(scope, receive, send) File "/usr/local/lib/python3.8/site-packages/starlette/routing.py", line 806, in app await self.default(scope, receive, send) File "/usr/local/lib/python3.8/site-packages/connexion/middleware/swagger_ui.py", line 235, in default_fn await self.app(original_scope, receive, send) File "/usr/local/lib/python3.8/site-packages/starlette/middleware/cors.py", line 85, in __call__ await self.app(scope, receive, send) File "/usr/local/lib/python3.8/site-packages/starlette/middleware/cors.py", line 85, in __call__ await self.app(scope, receive, send) File "/usr/local/lib/python3.8/site-packages/connexion/middleware/routing.py", line 154, in __call__ await self.router(scope, receive, send) File "/usr/local/lib/python3.8/site-packages/starlette/routing.py", line 756, in __call__ await self.middleware_stack(scope, receive, send) File "/usr/local/lib/python3.8/site-packages/starlette/routing.py", line 776, in app await route.handle(scope, receive, send) File "/usr/local/lib/python3.8/site-packages/starlette/routing.py", line 485, in handle await self.app(scope, receive, send) File "/usr/local/lib/python3.8/site-packages/starlette/routing.py", line 756, in __call__ await self.middleware_stack(scope, receive, send) File "/usr/local/lib/python3.8/site-packages/starlette/routing.py", line 776, in app await route.handle(scope, receive, send) File "/usr/local/lib/python3.8/site-packages/starlette/routing.py", line 297, in handle await self.app(scope, receive, send) File "/usr/local/lib/python3.8/site-packages/connexion/middleware/routing.py", line 48, in __call__ await self.next_app(original_scope, receive, send) File "/usr/local/lib/python3.8/site-packages/connexion/middleware/abstract.py", line 264, in __call__ return await operation(scope, receive, send) File "/usr/local/lib/python3.8/site-packages/connexion/middleware/security.py", line 106, in __call__ await self.next_app(scope, receive, send) File "/usr/local/lib/python3.8/site-packages/connexion/middleware/abstract.py", line 264, in __call__ return await operation(scope, receive, send) File "/usr/local/lib/python3.8/site-packages/connexion/middleware/request_validation.py", line 142, in __call__ await self.next_app(scope, receive, send) File "/usr/local/lib/python3.8/site-packages/connexion/middleware/abstract.py", line 264, in __call__ return await operation(scope, receive, send) File "/usr/local/lib/python3.8/site-packages/connexion/middleware/lifespan.py", line 26, in __call__ await self.next_app(scope, receive, send) File "/usr/local/lib/python3.8/site-packages/connexion/middleware/abstract.py", line 264, in __call__ return await operation(scope, receive, send) File "/usr/local/lib/python3.8/site-packages/connexion/middleware/context.py", line 25, in __call__ await self.next_app(scope, receive, send) File "/usr/local/lib/python3.8/site-packages/connexion/apps/flask.py", line 151, in __call__ return await self.asgi_app(scope, receive, send) File "/usr/local/lib/python3.8/site-packages/a2wsgi/wsgi.py", line 165, in __call__ return await responder(scope, receive, send) File "/usr/local/lib/python3.8/site-packages/a2wsgi/wsgi.py", line 200, in __call__ await self.loop.run_in_executor( File "/usr/local/lib/python3.8/concurrent/futures/thread.py", line 57, in run result = self.fn(*self.args, **self.kwargs) File "/usr/local/lib/python3.8/site-packages/a2wsgi/wsgi.py", line 256, in wsgi iterable = self.app(environ, start_response) File "/usr/local/lib/python3.8/site-packages/flask/app.py", line 2532, in wsgi_app response = self.handle_exception(e) File "/usr/local/lib/python3.8/site-packages/flask/app.py", line 2529, in wsgi_app response = self.full_dispatch_request() File "/usr/local/lib/python3.8/site-packages/flask/app.py", line 1825, in full_dispatch_request rv = self.handle_user_exception(e) File "/usr/local/lib/python3.8/site-packages/flask/app.py", line 1823, in full_dispatch_request rv = self.dispatch_request() File "/usr/local/lib/python3.8/site-packages/flask/app.py", line 1799, in dispatch_request return self.ensure_sync(self.view_functions[rule.endpoint])(**view_args) File "/usr/local/lib/python3.8/site-packages/connexion/apps/flask.py", line 68, in __call__ return self.fn(*args, **kwargs) File "/usr/local/lib/python3.8/site-packages/connexion/decorators/main.py", line 134, in wrapper return decorated_function(request) File "/usr/local/lib/python3.8/site-packages/connexion/decorators/response.py", line 171, in wrapper handler_response = function(*args, **kwargs) File "/usr/local/lib/python3.8/site-packages/connexion/decorators/parameter.py", line 87, in wrapper return function(**kwargs) File "/usr/local/lib/python3.8/site-packages/connexion/decorators/main.py", line 123, in wrapper return function(*args, **kwargs) File "/opt/airflow/airflow/api_connexion/security.py", line 182, in decorated return _requires_access( File "/opt/airflow/airflow/api_connexion/security.py", line 92, in _requires_access return func(*args, **kwargs) File "/opt/airflow/airflow/utils/session.py", line 79, in wrapper return func(*args, session=session, **kwargs) File "/opt/airflow/airflow/api_connexion/parameters.py", line 104, in wrapped_function return func(*args, **kwargs) TypeError: get_dataset_events() missing 1 required keyword-only argument: 'limit'

airflow/api_connexion/openapi/v1.yaml

potiuk · 2024-04-16T09:05:33Z

airflow/api_connexion/parameters.py

@@ -41,7 +41,7 @@ def validate_istimezone(value: datetime) -> None:
        raise BadRequest("Invalid datetime format", detail="Naive datetime is disallowed")


-def format_datetime(value: str) -> datetime:
+def format_datetime(value: str | None) -> datetime | None:


We needdd to convert the decorators to handle None as default value - see https://github.com/apache/airflow/pull/39055/files#r1566990719

potiuk · 2024-04-16T09:06:13Z

airflow/cli/commands/internal_api_command.py

-            debug=True,  # nosec
-            use_reloader=not app.config["TESTING"],
+            log_level="debug",
+            # reload=not app.app.config["TESTING"],


TODO: We still need to sort out this reload option here.

potiuk · 2024-04-16T09:06:54Z

airflow/cli/commands/internal_api_command.py

@@ -102,7 +102,7 @@ def internal_api(args):
            "--workers",
            str(num_workers),
            "--worker-class",
-            str(args.workerclass),
+            "uvicorn.workers.UvicornWorker",


TODO: We need to remove option of using different workers and only leave the uvicorn worker.

airflow/cli/commands/internal_api_command.py

potiuk · 2024-04-16T09:11:08Z

airflow/cli/commands/webserver_command.py

@@ -356,11 +356,11 @@ def webserver(args):
        print(f"Starting the web server on port {args.port} and host {args.hostname}.")
        app = create_app(testing=conf.getboolean("core", "unit_test_mode"))
        app.run(
-            debug=True,
-            use_reloader=not app.config["TESTING"],
+            log_level="debug",


TODO: We need to figure ouy our reloading here

airflow/www/extensions/init_views.py

Taragolis · 2024-04-16T09:06:32Z

tests/conftest.py

+@pytest.fixture(autouse=True)
+def create_swagger_ui_dir_if_missing():
+    """
+    The directory needs to exist to satisfy starlette attempting to register it as middleware
+    :return:
+    """
+    swagger_ui_dir = AIRFLOW_SOURCES_ROOT_DIR / "airflow" / "www" / "static" / "dist" / "swagger-ui"
+    swagger_ui_dir.mkdir(exist_ok=True, parents=True)


I think better propagate it in the CI and raise an error if "static" / "dist" not available, it might prevent some serious error rather than silence it

Good point.

airflow/cli/commands/internal_api_command.py

airflow/cli/commands/webserver_command.py

airflow/providers/fab/auth_manager/fab_auth_manager.py

potiuk · 2024-04-16T09:15:51Z

airflow/www/app.py

-    flask_app = Flask(__name__)
+    connexion_app = connexion.FlaskApp(__name__)
+
+    @connexion_app.app.before_request


Here the question is if that exemption here from CSRF is teh right one

TODO: also add FAB cc: @vincbeck ?

Correct. The path is /auth/fab/v1. Ideally we should have a set containing all the paths that need to ignore CSRF. SO far I can see only 2: Airflow API and Fab API

Yes. If we go that direction we should likely propose a better way than it is currently (it's a bit messy).

airflow/www/app.py

potiuk · 2024-04-16T09:17:57Z

airflow/www/extensions/init_appbuilder_links.py

@@ -53,7 +53,7 @@ def init_appbuilder_links(app):
        appbuilder.add_link(
            name=RESOURCE_DOCS,
            label="REST API Reference (Swagger UI)",
-            href="/api/v1./api/v1_swagger_ui_index",
+            href="/api/v1/ui",


We use embedded swagger UI now.

potiuk · 2024-04-16T09:19:39Z

airflow/www/extensions/init_views.py

+    auth_mgr.set_api_endpoints(connexion_app)
+
+
+def init_cors_middleware(connexion_app: connexion.FlaskApp):


@RobbeSneyders - followint your advice we replaced our custom CORS implementation with the middleware - does it look right?

airflow/www/extensions/init_views.py

potiuk · 2024-04-16T09:21:59Z

airflow/www/extensions/init_views.py

-    app.register_error_handler(404, views.not_found)
-
-
-def set_cors_headers_on_response(response):


Replaced by CORS middleware.

potiuk · 2024-04-16T09:24:33Z

newsfragments/37638.significant.rst

@@ -0,0 +1,4 @@
+Replaced test_should_respond_400_on_invalid_request with test_ignore_read_only_fields in the test_dag_endpoint.py.


TODO: We need to add a few more notes likely here and change the newsfragment number to reflect the PR number.

potiuk · 2024-04-18T11:46:51Z

@RobbeSneyders

I have to admit that the usage of Connexion by Airflow was a bit of a blind spot for us. Connexion was used differently here than we anticipated, leading to a much larger migration effort than for most of our users.

Indeed. It's a huge one.

Aligning the implementation more with our intended interfaces should improve the robustness towards the future though. And the redesign of Connexion 3 makes Connexion itself more future-proof and easier to maintain as well. So there is definitely long-term value in this migration.

Yes. I truly hope we can make the decision eventually that we want to go this direction - seeing a number of changes and potential disruptions it may cause makes - of course - maintainers concerned. After trying it out and seeing what it takes, it's not relly a simple "library version" upgrade. So it's not likely the PR will make it into the codebase in this form. But I hope we can build a consensus and a task force to make it happen - so any comments and insights on what else we can do to make it more "standard" is really appreciated. I started a discussion - where I hope we can make at least decision on moving things in the right direction: https://lists.apache.org/thread/yrvp2mg25xcwznt8yr9dmdmxrmomwjs2

Now that we understand better how Airflow uses Connexion, we will keep it in mind when taking future decisions and we hope that you will keep actively providing feedback. We would rather collaborate on improvements upstream than having you depend on workarounds.

Yes. If you can see something that could be improved and make our lives better - it will still take quite some time to make this one into "merged" status, so if there is anything Connexion maintainers could do to address the use case we have and make it esier and more future-proof to migrate it would be great.

I already addressed a few of your comments, but need some more time to look into the remaining ones.

Please. I think a number of our decisions were mostly guesses, and even the way we have been integrating in the past were somewhat not the way things were intended originally, so any comments and insights from your side would be great.

potiuk · 2024-04-20T17:01:44Z

airflow/www/app.py

+
+    init_cors_middleware(conn_app)
+
+    flask_app = conn_app.app


TODO: init_wsgl_middleware/proxy_fix does not work currently -> BASE_URL does not work.

This is a huge PR being result of over a 100 commits made by a number of people in #apache#36052 and apache#37638. It switches to Connexion 3 as the driving backend implementation for both - Airflow REST APIs and Flask app that powers Airflow UI. It should be largely backwards compatible when it comes to behaviour of both APIs and Airflow Webserver views, however due to decisions made by Connexion 3 maintainers, it changes heavily the technology stack used under-the-hood: 1) Connexion App is an ASGI-compatible Open-API spec-first framework using ASGI as an interface between webserver and Python web application. ASGI is an asynchronous successor of WSGI. 2) Connexion itself is using Starlette to run asynchronous web services in Python. 3) We continue using gunicorn appliation server that still uses WSGI standard, which means that we can continue using Flask and we are usig standard Uvicorn ASGI webserver that converts the ASGI interface to WSGI interface of Gunicorn Some of the problems handled in this PR There were two problem was with session handling: * the get_session_cookie - did not get the right cookie - it returned "session" string. The right fix was to change cookie_jar into cookie.jar because this is where apparently TestClient of starlette is holding the cookies (visible when you debug) * The client does not accept "set_cookie" method - it accepts passing cookies via "cookies" dictionary - this is the usual httpx client - see https://www.starlette.io/testclient/ - so we have to set cookie directly in the get method to try it out Add "flask_client_with_login" for tests that neeed flask client Some tests require functionality not available to Starlette test client as they use Flask test client specific features - for those we have an option to get flask test client instead of starlette one. Fix error handling for new connection 3 approach Error handling for Connexion 3 integration needed to be reworked. The way it behaves is much the same as it works in main: * for API errors - we get application/problem+json responses * for UI erros - we have rendered views * for redirection - we have correct location header (it's been missing) * the api error handled was not added as available middleware in the www tests It should fix all test_views_base.py tests which were failing on lack of location header for redirection. Fix wrong response is tests_view_cluster_activity The problem in the test was that Starlette Test Client opens a new connection and start new session, while flask test client uses the same database session. The test did not show data because the data was not committed and session was not closed - which also failed sqlite local tests with "database is locked" error. Fix test_extra_links The tests were failing again because the dagrun created was not committed and session not closed. This worked with flask client that used the same session accidentally but did not work with test client from Starlette. Also it caused "database locked" in sqlite / local tests. Switch to non-deprecated auth manager Fix to test_views_log.py This PR partially fixes sessions and request parameter for test_views_log. Some tests are still failing but for different reasons - to be investigated. Fix views_custom_user_views tests The problem in those tests was that the check in security manager was based on the assumption that the security manager was shared between the client and test flask application - because they were coming from the same flask app. But when we use starlette, the call goes to a new process started and the user is deleted in the database - so the shortcut of checking the security manager did not work. The change is that we are now checking if the user is deleted by calling /users/show (we need a new users READ permission for that) - this way we go to the database and check if the user was indeed deleted. Fix test_task_instance_endpoint tests There were two reasons for the test failed: * when the Job was added to task instance, the task instance was not merged in session, which means that commit did not store the added Job * some of the tests were expecting a call with specific session and they failed because session was different. Replacing the session with mock.ANY tells pytest that this parameter can be anything - we will have different session when when the call will be made with ASGI/Starlette Fix parameter validation * added default value for limit parameter across the board. Connexion 3 does not like if the parameter had no default and we had not provided one - even if our custom decorated was adding it. Adding default value and updating our decorator to treat None as `default` fixed a number of problems where limits were not passed * swapped openapi specification for /datasets/{uri} and /dataset/events. Since `{uri}` was defined first, connection matched `events` with `{uri}` and chose parameter definitions from `{uri}` not events Fix test_log_enpoint tests The problem here was that some sessions should be committed/closed but also in order to run it standalone we wanted to create log templates in the database - as it relied implcitly on log templates created by other tests. Fix test_views_dagrun, test_views_tasks and test_views_log Fixed by switching to use flask client for testing rather than starlette. Starlette client in this case has some side effects that are also impacting Sqlite's session being created in a different thread and deleted with close_all_sessions fixture. Fix test_views_dagrun Fixed by switching to use flask client for testing rather than starlette. Starlette client in this case has some side effects that are also impacting Sqlite's session being created in a different thread and deleted with close_all_sessions fixture. Co-authored-by: sudipto baral <[email protected]> Co-authored-by: satoshi-sh <[email protected]> Co-authored-by: Maksim Yermakou <[email protected]> Co-authored-by: Ulada Zakharava <[email protected]> Better API initialization including vending of API specification. The way paths are added and initialized is better (for example FAB contributes their path via new method in Auth Manager. This also add back-compatibility to FAB auth manaager to continue working on Airflow 2.9.

github-actions · 2024-08-29T00:14:03Z

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.

boring-cyborg bot added area:API Airflow's REST/HTTP API area:CLI area:db-migrations PRs with DB migration area:dev-tools area:providers area:UI Related to UI/UX. For Frontend Developers. area:webserver Webserver related Issues kind:documentation provider:fab labels Apr 16, 2024

potiuk mentioned this pull request Apr 16, 2024

Migrate to connexion v3 #37638

Closed

7 tasks